package cn.me.alphamq.service.impl;

import cn.me.alphamq.client.DefaultConsole;
import cn.me.alphamq.common.entity.resp.ListAllTopicInfoResp;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.common.model.broker.ConsumeQueueFileModel;
import cn.me.alphamq.common.util.AssertUtil;
import cn.me.alphamq.service.ConsoleService;
import cn.me.alphamq.vo.ConsumerOffsetInfoVO;
import cn.me.alphamq.vo.PageVO;
import cn.me.alphamq.vo.Response;
import cn.me.alphamq.vo.TopicInfoVO;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * @author f
 */
@Service
public class ConsoleServiceImpl implements ConsoleService {

    private final DefaultConsole defaultConsole;

    public ConsoleServiceImpl(DefaultConsole defaultConsole) {
        this.defaultConsole = defaultConsole;
    }

    @Override
    public Response pageQueryTopicInfo(String topic, Integer page, Integer size) {
        AssertUtil.isTrue(size <= 100, "Page size must be less than 100!");

        List<TopicInfoVO> topicInfoVOList = defaultConsole.sendListAllTopicInfoMsg().getCommitLogTopicModelList()
                .stream().map(TopicInfoVO::new).collect(Collectors.toList());

        if (topic != null && !topic.isEmpty()) {
            topicInfoVOList = topicInfoVOList.stream().filter(topicInfoVO ->
                    topicInfoVO.getTopic().contains(topic)).collect(Collectors.toList());
        }

        int total = topicInfoVOList.size();
        int fromIndex = Math.min((page - 1) * size, total);
        int toIndex = Math.min(page * size, total);

        return Response.ok(new PageVO<>(
                topicInfoVOList.subList(fromIndex, toIndex), page,
                (int) Math.ceil((double) total / size)
        ));
    }

    @Override
    public Response listConsumerOffsetInfo(String topic) {
        AssertUtil.isNotBlank(topic, "Topic is blank!");

        ListAllTopicInfoResp listAllTopicInfoResp = defaultConsole.sendListAllTopicInfoMsg();

        Optional<CommitLogTopicModel> commitLogTopicModelOp = listAllTopicInfoResp.getCommitLogTopicModelList()
                .stream().filter(topicModel -> topic.equals(topicModel.getTopic()))
                .findAny();
        if (!commitLogTopicModelOp.isPresent()) {
            return Response.fail();
        }
        CommitLogTopicModel commitLogTopicModel = commitLogTopicModelOp.get();
        long createAt = commitLogTopicModel.getCreateAt();
        long updateAt = commitLogTopicModel.getUpdateAt();
        List<ConsumeQueueFileModel> queueList = commitLogTopicModel.getQueueList();

        Map<String, Map<String, String>> consumerGroupMap = listAllTopicInfoResp.getConsumeQueueOffsetModel()
                .fetchConsumerGroupMapForcibly(topic);

        List<ConsumerOffsetInfoVO> consumerOffsetInfoVOList = new ArrayList<>();
        consumerGroupMap.forEach((consumerGroup, consumeQueueMap) -> {
            consumeQueueMap.forEach((queueIdStr, consumeDetail) -> {
                String[] split = consumeDetail.split("#");
                int consumeQueueOffset = Integer.parseInt(split[1]);
                int queueId = Integer.parseInt(queueIdStr);

                consumerOffsetInfoVOList.add(new ConsumerOffsetInfoVO(
                        queueId, consumerGroup, consumeQueueOffset,
                        queueList.get(queueId).getOffsetLimit(),
                        createAt, updateAt
                ));
            });
        });

        return Response.ok(consumerOffsetInfoVOList);
    }
}
